package pers.cz.logtail.queue.kafka;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import pers.cz.logtail.queue.Message;
import pers.cz.logtail.queue.Producer;
import pers.cz.logtail.queue.QueueConf;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @program: JefLogTail
 * @description: 生产者
 * @author: Cheng Zhi
 * @create: 2023-03-15 17:04
 **/
public class KafkaProducer implements Producer {

    private Logger logger = LoggerFactory.getLogger(KafkaProducer.class);
    org.apache.kafka.clients.producer.KafkaProducer<String,String> kafkaProducer;

    public KafkaProducer(QueueConf queueConf) {
        this.kafkaProducer = new org.apache.kafka.clients.producer.KafkaProducer<String, String>(initConfig(queueConf));
    }
    private Map<String, Object> initConfig(QueueConf queueConf){

        Map<String, Object> prop = new HashMap<>();

        prop.put("bootstrap.servers", queueConf.getAddstr());
        prop.put("key.serializer", queueConf.getKeySerializer());
        prop.put("value.serializer", queueConf.getValueSerializer());
        prop.put("acks", queueConf.getAcks());
        prop.put("retries", queueConf.getRetries());
        prop.put("batch.size", queueConf.getBatchSize());
        prop.put("linger.ms", queueConf.getLingerMs());
        prop.put("buffer.memory", queueConf.getBufferMemory());
        return prop;
    }

    /**
     * 同步发送
     * @param ms
     * @return
     */
    @Override
    public Message sendSync(Message ms) {
        Future<RecordMetadata> task = kafkaProducer
                .send(new ProducerRecord<String, String>(ms.getTopic(), null, ms.getValueStr()));
        try {
            task.get();
        } catch (InterruptedException e) {
            logger.error(e.getMessage());
        } catch (ExecutionException e) {
            logger.error(e.getMessage());
        }
        return ms;
    }

    /**
     * 异步发送
     * @param ms
     * @return
     */
    @Override
    public Message sendAsync(Message ms) {
        kafkaProducer.send(new ProducerRecord<String, String>(ms.getTopic(),null,ms.getValueStr()));
        return ms;
    }

    @Override
    public void close() {
        kafkaProducer.close();
    }
}
